package com.sunzm.spark.sql.hive.calllog

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.DateFormatUtils
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}

/**
 *
 * 通过SparkSQL操作Hive
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-07-06 17:54
 */
object SparkSqlHiveCallLog {
  def main(args: Array[String]): Unit = {

    //这句话需要在 SparkSession 创建之前加
    System.setProperty("HADOOP_USER_NAME", "hdfs")

    val warehouseLocation = "/user/hive/warehouse"
    //val warehouseLocation = new File("spark-warehouse").getAbsolutePath

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[*]")
      .config("spark.default.parallelism", 8)
      .config("spark.sql.shuffle.partitions", 8)
      //如果集群中有hive相关的配置文件，下面的都可以不写
      //.config("fs.defaultFS", "hdfs://192.168.1.158:8020")
      .config("fs.defaultFS", "file:///")
      .config("spark.sql.warehouse.dir", warehouseLocation)
      //如果不开启对Hive的支持,元数据保存在内存中,程序退出,元数据也就丢了
      .enableHiveSupport()
      //开启动态分区支持
      .config("hive.exec.dynamic.partition", "true")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("javax.jdo.option.ConnectionDriverName", "com.mysql.jdbc.Driver")
      .config("javax.jdo.option.ConnectionURL", "jdbc:mysql://82.156.210.70:3306/hive?useSSL=false")
      .config("javax.jdo.option.ConnectionUserName", "root")
      .config("javax.jdo.option.ConnectionPassword", "ABC123abc.123")
      //高版本的Hive需要设置下面的参数
      .config("hive.metastore.schema.verification", false)
      .config("datanucleus.schema.autoCreateTables", true)
      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._

    //在 call_db 库中建一个表
    spark.sql("USE call_db")

    val resutDate = "2020-11-15"

    val dataDS: Dataset[String] = spark.read.textFile(s"/data/data_center/cus/call_log_record/call_log_record-${resutDate}*")

    val callRecordDS: Dataset[CallRecord] = dataDS.filter(line => {
      var res = false
      try {
        val jSONObject = JSON.parseObject(line)
        val callRecordId = jSONObject.getString("callRecordId")
        if (StringUtils.isNotBlank(callRecordId)) {
          res = true
        }
      } catch {
        case e: Throwable => e.printStackTrace()
      }
      res
    }).map(line => {
      val jSONObject = JSON.parseObject(line)
      val callRecordId: String = jSONObject.getString("callRecordId")
      val parentCallRecordId: String = jSONObject.getString("parentCallRecordId")
      val companyId: String = jSONObject.getString("companyId")
      val appId: String = jSONObject.getString("appId")
      val voiceAliyunUrl: String = jSONObject.getString("voiceAliyunUrl")
      val numberProvice: String = jSONObject.getString("numberProvice")
      val numberCity: String = jSONObject.getString("numberCity")
      val receiveStaffNames: String = jSONObject.getString("receiveStaffNames")
      val feeDuration: Int = jSONObject.getIntValue("feeDuration")
      val callResult: Int = jSONObject.getIntValue("callResult")
      val callFlag: Int = jSONObject.getIntValue("callFlag")
      val direction: Int = jSONObject.getIntValue("direction")
      val duration: Long = jSONObject.getLongValue("duration")
      val intervalDuration: Long = jSONObject.getLongValue("intervalDuration")
      val callDirection: Long = jSONObject.getLongValue("callDirection")
      val startTime: Long = jSONObject.getLongValue("startTime")
      val endTime: Long = jSONObject.getLongValue("endTime")

      val result_date = DateFormatUtils.format(startTime, "yyyy-MM-dd")

      val userAnswerTime: Long = jSONObject.getLongValue("userAnswerTime")
      val agentStartTime: Long = jSONObject.getLongValue("agentStartTime")

      CallRecord(callRecordId, parentCallRecordId, companyId, appId, voiceAliyunUrl, numberProvice, numberCity,
        receiveStaffNames, feeDuration, callResult, callFlag, direction, duration, intervalDuration, callDirection,
        startTime, endTime, userAnswerTime, agentStartTime, result_date)
    })

    //使用DSL风格API插入Hive的表中
    //注意插入的字段个数，不能多于表中实际的字段个数
   /* callRecordDS
      .select("callRecordId", "parentCallRecordId", "appId", "voiceAliyunUrl", "result_date")
      .write
      .format("hive")
      .mode(SaveMode.Append)
      .partitionBy("result_date")
      .saveAsTable("call_log_record")*/


    //使用SQL语句插入Hive的表中
    callRecordDS.createOrReplaceTempView("v_call_record")
    spark.sql(
      """
        |  INSERT INTO TABLE call_log_record
        |     partition (result_date)
        |  SELECT
        |     callRecordId,
        |     parentCallRecordId,
        |     appId,
        |     voiceAliyunUrl,
        |     result_date
        |   FROM v_call_record
        |""".stripMargin)

    //语法：LOAD DATA [ LOCAL ] INPATH path [ OVERWRITE ] INTO TABLE table_identifier [ partition_spec ]
   /* spark.sql(
      """
        |LOAD DATA INPATH '/user/hive/warehouse/test_partition/c2=2/c3=3'
        |    OVERWRITE INTO TABLE test_load_partition PARTITION (c2=2, c3=3);
        |""".stripMargin)*/


    spark.sql("show partitions call_log_record").show(10, false)

    spark.sql(s"SELECT * FROM call_log_record WHERE result_date = '${resutDate}' LIMIT 10")
      .show(10, false)

    spark.stop()
  }

  /**
   * 封装通话记录的样例类
   *
   * @param callRecordId
   * @param parentCallRecordId
   * @param companyId
   * @param appId
   * @param voiceAliyunUrl
   * @param numberProvice
   * @param numberCity
   * @param receiveStaffNames
   * @param feeDuration
   * @param callResult
   * @param callFlag
   * @param direction
   * @param duration
   * @param intervalDuration
   * @param callDirection
   * @param startTime
   * @param endTime
   * @param userAnswerTime
   * @param agentStartTime
   * @param result_date
   */
  private case class CallRecord(callRecordId: String,
                                parentCallRecordId: String,
                                companyId: String,
                                appId: String,
                                voiceAliyunUrl: String,
                                numberProvice: String,
                                numberCity: String,
                                receiveStaffNames: String,
                                feeDuration: Int,
                                callResult: Int,
                                callFlag: Int,
                                direction: Int,
                                duration: Long,
                                intervalDuration: Long,
                                callDirection: Long,
                                startTime: Long,
                                endTime: Long,
                                userAnswerTime: Long,
                                agentStartTime: Long,
                                result_date: String)

}
